package com.ververica.cdc.connectors.mysql.testutils;

import java.net.URI;
import java.net.URL;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import static org.junit.Assert.assertNotNull;

/**
 * Creates and populates a unique instance of a MySQL database for each run of a JUnit test.
 * Users of this class need to provide a JDBC URL, logical database name, username, and password.
 * It is expected that there is an initialization SQL file in
 * <code>src/test/resources/ddl/&lt;database_name&gt;.sql</code>. The database name is enriched
 * with a unique suffix that guarantees complete isolation between test runs, resulting in a
 * database name like <code>&lt;database_name&gt;_&lt;suffix&gt;</code>.
 *
 * <p>This class is inspired by the Debezium project.
 */

public class UniqueDatabase {

    // ------------------------------------------------------------------------
    //  Static SQL templates
    // ------------------------------------------------------------------------

    private static final String[] CREATE_DATABASE_DDL =
            new String[] {"CREATE DATABASE `$DBNAME$`;", "USE `$DBNAME$`;"};
    private static final String DROP_DATABASE_DDL = "DROP DATABASE IF EXISTS `$DBNAME$`;";
    private static final Pattern COMMENT_PATTERN = Pattern.compile("^(.*)--.*$");

    // ------------------------------------------------------------------------
    //  Configuration fields
    // ------------------------------------------------------------------------

    /** The base JDBC URL (without the specific database appended). */
    private final String baseJdbcUrl;

    /** The final unique database name (including the random suffix). */
    private final String databaseName;

    /** The template (logical) database name from which we build the final name. */
    private final String templateName;

    /** The username for MySQL authentication. */
    private final String username;

    /** The password for MySQL authentication. */
    private final String password;

    /** Host extracted from the JDBC URL. */
    private final String host;

    /** Port extracted from the JDBC URL. */
    private final int port;

    // ------------------------------------------------------------------------
    //  Constructors
    // ------------------------------------------------------------------------

    /**
     * Constructs a UniqueDatabase instance with a randomly generated suffix appended
     * to the provided {@code databaseName}.
     *
     * @param jdbcUrl      The base JDBC URL to connect to MySQL (e.g., {@code jdbc:mysql://localhost:3306})
     * @param databaseName A logical name used to locate the init script and form the final database name
     * @param username     The username for MySQL authentication
     * @param password     The password for MySQL authentication
     */
    public UniqueDatabase(String jdbcUrl, String databaseName, String username, String password) {
        this(
                jdbcUrl,
                databaseName,
                Integer.toUnsignedString(new Random().nextInt(), 36),
                username,
                password
        );
    }

    /**
     * Private constructor that appends a unique {@code identifier} to the database name.
     */
    private UniqueDatabase(
            String jdbcUrl,
            String databaseName,
            String identifier,
            String username,
            String password) {
        jdbcUrl = String.format(jdbcUrl, databaseName);
        // Parse host and port from the JDBC URL
        ParsedUrl parsedUrl = parseHostAndPort(jdbcUrl);
        this.host = parsedUrl.host;
        this.port = parsedUrl.port;

        // Remove any trailing database portion from the base URL
        this.baseJdbcUrl = removeDatabaseFromJdbcUrl(jdbcUrl);

        // Build the unique final database name
        this.databaseName = databaseName + "_" + identifier;
        this.templateName = databaseName;
        this.username = username;
        this.password = password;
    }

    // ------------------------------------------------------------------------
    //  Public API
    // ------------------------------------------------------------------------

    /**
     * Returns the extracted host from the JDBC URL.
     */
    public String getHost() {
        return host;
    }

    /**
     * Returns the extracted port from the JDBC URL.
     */
    public int getPort() {
        return port;
    }

    /**
     * Returns the base JDBC URL (without the final database appended).
     */
    public String getBaseJdbcUrl() {
        return baseJdbcUrl;
    }

    /**
     * Returns the unique database name.
     */
    public String getDatabaseName() {
        return databaseName;
    }

    /**
     * Returns the username for MySQL authentication.
     */
    public String getUsername() {
        return username;
    }

    /**
     * Returns the password for MySQL authentication.
     */
    public String getPassword() {
        return password;
    }

    /**
     * Returns a fully qualified table name in the format
     * {@code <databaseName>.<tableName>}.
     */
    public String qualifiedTableName(final String tableName) {
        return String.format("%s.%s", databaseName, tableName);
    }

    /**
     * Creates the unique database and initializes it with the script located in
     * {@code src/test/resources/ddl/<databaseName>.sql}.
     */
    public void createAndInitialize() {
        final String ddlFile = String.format("ddl/%s.sql", templateName);
        final URL ddlTestFile = UniqueDatabase.class.getClassLoader().getResource(ddlFile);
        assertNotNull("Cannot locate " + ddlFile, ddlTestFile);

        try (Connection connection = DriverManager.getConnection(baseJdbcUrl, username, password);
             Statement statement = connection.createStatement()) {

            // 1) Create and USE the new database
            for (String ddl : CREATE_DATABASE_DDL) {
                statement.execute(convertSQL(ddl));
            }

            // 2) Read and execute the initialization SQL script
            List<String> sqlLines = Files.readAllLines(Paths.get(ddlTestFile.toURI()));
            String sqlContent = sqlLines.stream()
                    .map(String::trim)
                    .filter(line -> !line.startsWith("--") && !line.isEmpty())
                    .map(line -> {
                        Matcher matcher = COMMENT_PATTERN.matcher(line);
                        return matcher.matches() ? matcher.group(1) : line;
                    })
                    .collect(Collectors.joining(" "));

            // Split the script into statements by ';'
            List<String> statementsList = Arrays.stream(sqlContent.split(";"))
                    .map(String::trim)
                    .filter(stmt -> !stmt.isEmpty())
                    .collect(Collectors.toList());

            for (String stmt : statementsList) {
                statement.execute(stmt);
            }
        } catch (Exception e) {
            throw new IllegalStateException("Failed to create and initialize the database.", e);
        }
    }

    /**
     * Drops the unique database if it exists.
     */
    public void dropDatabase() {
        try (Connection connection = DriverManager.getConnection(baseJdbcUrl, username, password);
             Statement statement = connection.createStatement()) {
            statement.execute(convertSQL(DROP_DATABASE_DDL));
        } catch (Exception e) {
            throw new IllegalStateException("Failed to drop the database.", e);
        }
    }

    /**
     * Returns a JDBC connection specifically to the unique database.
     */
    public Connection getJdbcConnection() throws SQLException {
        String dbSpecificUrl = appendDatabaseToJdbcUrl(baseJdbcUrl, databaseName);
        return DriverManager.getConnection(dbSpecificUrl, username, password);
    }

    // ------------------------------------------------------------------------
    //  Internal utilities
    // ------------------------------------------------------------------------

    /**
     * Replaces the {@code $DBNAME$} placeholder with the actual database name.
     */
    private String convertSQL(final String sql) {
        return sql.replace("$DBNAME$", databaseName);
    }

    /**
     * Removes any trailing {@code /<dbName>} part from the JDBC URL,
     * returning a base URL that ends at host:port.
     * <p>
     * Example:
     * <ul>
     *   <li>{@code jdbc:mysql://localhost:3306/db} -> {@code jdbc:mysql://localhost:3306}</li>
     *   <li>{@code jdbc:mysql://localhost:3306}     -> {@code jdbc:mysql://localhost:3306}</li>
     * </ul>
     */
    private String removeDatabaseFromJdbcUrl(String jdbcUrl) {
        ParsedUrl parsedUrl = parseHostAndPort(jdbcUrl);
        // We reconstruct the base URL without a path, so it ends at host:port
        return "jdbc:mysql://" + parsedUrl.host + ":" + parsedUrl.port;
    }

    /**
     * Appends the given {@code dbName} to the base JDBC URL.
     * <p>
     * Example:
     * <ul>
     *   <li>{@code jdbc:mysql://localhost:3306} + {@code mydb} -> {@code jdbc:mysql://localhost:3306/mydb}</li>
     * </ul>
     */
    private String appendDatabaseToJdbcUrl(String baseUrl, String dbName) {
        return baseUrl.endsWith("/") ? baseUrl + dbName : baseUrl + "/" + dbName;
    }

    /**
     * Parses the host and port from a JDBC URL of the form:
     * <pre>{@code
     *   jdbc:mysql://hostname:port/optional_database
     * }</pre>
     * <p>If the port is not specified, defaults to {@code 3306}.
     */
    private ParsedUrl parseHostAndPort(String jdbcUrl) {
        // Remove the "jdbc:" prefix
        // e.g. "jdbc:mysql://localhost:3306/db" -> "mysql://localhost:3306/db"
        String withoutJdbc = jdbcUrl.substring("jdbc:".length());
        URI uri = URI.create(withoutJdbc); // "mysql://localhost:3306/db"

        String host = uri.getHost();
        int port = uri.getPort() == -1 ? 3306 : uri.getPort();

        if (host == null) {
            throw new IllegalArgumentException("Unable to parse host from JDBC URL: " + jdbcUrl);
        }
        return new ParsedUrl(host, port);
    }

    /**
     * Simple container for the parsed host and port.
     */
    private static class ParsedUrl {
        final String host;
        final int port;

        ParsedUrl(String host, int port) {
            this.host = host;
            this.port = port;
        }
    }
}
